﻿using Infrastructure.ElasticSearch.Models;
using Infrastructure.ElasticSearch.ParentChilds;
using Microsoft.Extensions.Logging;
using Nest;
using Newtonsoft.Json;
using Newtonsoft.Json.Converters;
using System;
using System.Collections.Generic;
using System.Dynamic;
using System.Linq;
using System.Threading.Tasks;

namespace Infrastructure.ElasticSearch.Repository
{
    public abstract class EsRepositoryBase<TAggregateRoot> : IEsRepositoryBase<TAggregateRoot> where TAggregateRoot : EsAggregateRoot
    {

        private readonly string _hostEnvironmentName;
        private readonly IElasticSearchClientProvider _provider;
        private readonly Lazy<IElasticClient> esLazyClient;
        protected ILogger<EsRepositoryBase<TAggregateRoot>> _logger;

        protected EsRepositoryBase(IElasticSearchClientProvider clientProvider, ILogger<EsRepositoryBase<TAggregateRoot>> logger)
        {
            _provider = clientProvider;
            _hostEnvironmentName = clientProvider.HostEnvironmentName;
            esLazyClient = new Lazy<IElasticClient>(() => _provider.CreateElasticClient(GetEnvIndexName()));
        }

        private IElasticClient _esClient => esLazyClient.Value;

        public string GetEnvIndexName()
        {
            var indexName = GetIndexName();
            if (string.IsNullOrEmpty(indexName))
            {
                throw new Exception("未配置索引名称！");
            }

            var envName = GetEnvName();
            envName = !string.IsNullOrEmpty(envName) ? envName : _hostEnvironmentName;

            return $"{indexName}{envName}";
        }

        public abstract string GetIndexName();
        public abstract string GetEnvName();
        public async Task CreateIndexAsync(int shard = 5, int numberOfReplicas = 1)
        {
            var indexName = GetEnvIndexName();
            var exits = await _esClient.Indices.ExistsAsync(indexName);

            if (exits.Exists) return;

            var result = await _esClient
                .Indices.CreateAsync(indexName,
                    s =>
                        s.Index(indexName).Settings(o =>
                                o.NumberOfShards(shard)
                                    .NumberOfReplicas(numberOfReplicas)
                                    .Setting("max_result_window", int.MaxValue))
                            .Map(m => m.AutoMap<TAggregateRoot>().Dynamic(false)));

            if (result.Acknowledged)
            {
                return;
            }

            throw new ElasticSearchException($"Create Index {indexName}，failed : :" + result.ServerError.Error.Reason);
        }

        public async Task CreateIndexAsync(Func<TypeMappingDescriptor<object>, ITypeMapping> selector, int shard = 1, int numberOfReplicas = 1)
        {
            var indexName = GetEnvIndexName();
            var exits = await _esClient.Indices.ExistsAsync(indexName);

            if (exits.Exists) return;

            var result = await _esClient
                .Indices.CreateAsync(indexName,
                    ss =>
                        ss.Index(indexName)
                            .Settings(
                                o => o.NumberOfShards(shard).NumberOfReplicas(numberOfReplicas)
                                    .Setting("max_result_window", int.MaxValue))
                            .Map(selector));

            if (result.Acknowledged)
            {
                return;
            }

            throw new ElasticSearchException($"Create Index {indexName}，failed :" + result.ServerError.Error.Reason);
        }

        public async Task AddDocumentAsync(TAggregateRoot model, string routing = "")
        {
            var indexName = GetEnvIndexName();

            var result = await _esClient.IndexAsync<TAggregateRoot>(model, ss =>
                !string.IsNullOrEmpty(routing) ? ss.Index(indexName).Routing(routing) : ss.Index(indexName));

            if (result.ServerError == null) return;
            throw new ElasticSearchException($"Insert Document failed at index {indexName} :" + result.ServerError.Error.Reason);
        }

        public async Task AddOrUpdateDocumentAsync(TAggregateRoot model, string routing = "")
        {
            var indexName = GetEnvIndexName();
            var exits = await _esClient.DocumentExistsAsync(DocumentPath<TAggregateRoot>.Id(new Id(model)), dd => dd.Index(indexName));

            if (exits.Exists)
            {
                var result = await _esClient.UpdateAsync(DocumentPath<TAggregateRoot>.Id(new Id(model)),
                    ss =>
                        !string.IsNullOrEmpty(routing)
                            ? ss.Index(indexName).Routing(routing).Doc(model).RetryOnConflict(3)
                            : ss.Index(indexName).Doc(model).RetryOnConflict(3));

                if (result.ServerError == null) return;
                throw new ElasticSearchException($"Update Document failed at index{indexName} :" + result.ServerError.Error.Reason);
            }
            else
            {
                var result = await _esClient.IndexAsync<TAggregateRoot>(model, ss =>
                    !string.IsNullOrEmpty(routing) ? ss.Index(indexName).Routing(routing) : ss.Index(indexName));

                if (result.ServerError == null) return;
                throw new ElasticSearchException($"Insert Document failed at index {indexName} :" + result.ServerError.Error.Reason);
            }
        }

        public async Task DeleteManyAsync(List<TAggregateRoot> list, int bulkNum = 1000)
        {
            var indexName = GetEnvIndexName();
            if (list.Count <= bulkNum)
            {
                var response = await _esClient.DeleteManyAsync(list, indexName);
                if (response.Errors)
                    throw new ElasticSearchException($"Bulk DeleteMany Document failed at index {indexName} :{response.ServerError.Error.Reason}");
            }
            else
            {
                var total = (int)Math.Ceiling(list.Count * 1.0f / bulkNum);
                var tasks = new List<Task>();
                for (var i = 0; i < total; i++)
                {
                    var j = i;
                    tasks.Add(Task.Factory.StartNew(() => _esClient.DeleteManyAsync(list.Skip(j * bulkNum).Take(bulkNum).ToList(), indexName)));
                }
                await Task.WhenAll(tasks);
            }
        }

        public async Task PartialUpdateDocumentAsync(string index, TAggregateRoot model)
        {
            var json = model.ToJsonByElasticSearch();
            //var converter = new ExpandoObjectConverter();
            dynamic expandoObject = JsonConvert.DeserializeObject<ExpandoObject>(json, new ExpandoObjectConverter());
            var indexName = GetEnvIndexName();
            var response = await _esClient.UpdateAsync<TAggregateRoot, object>(index, s => s.Index(indexName).Doc(expandoObject));
            if (response.ServerError == null) return;
            throw new ElasticSearchException($"Bulk PartialUpdateDocumentAsync Document failed at index {indexName} :{response.ServerError.Error.Reason}");
        }

        public async Task PartialUpdateDocumentByNestedAsync(string index, string script, Dictionary<string, object> scriptParams)
        {
            var indexName = GetEnvIndexName();
            var response = await _esClient.UpdateAsync<TAggregateRoot, object>(index,
                s =>
                    s.Index(indexName).Script(ss => ss.Source(script).Params(scriptParams)));

            if (response.ServerError == null) return;
            throw new ElasticSearchException($"Bulk PartialUpdateDocumentAsync Document failed at index {indexName} :{response.ServerError.Error.Reason}");
        }

        public async Task<string> BulkAddOrUpdateDocumentAsync(List<TAggregateRoot> list, int bulkNum = 5000, string indexName = "")
        {
            if (indexName.IsNullOrWhiteSpace())
            {
                indexName = GetEnvIndexName();
            }
            var result = string.Empty;

            if (list.Count <= bulkNum)
            {
                result = await BulkAddOrUpdate(indexName, list);
            }
            else
            {
                var total = (int)Math.Ceiling(list.Count * 1.0f / bulkNum);
                var tasks = new List<Task>();
                for (var i = 0; i < total; i++)
                {
                    var j = i;
                    tasks.Add(Task.Factory.StartNew(() => result += BulkAddOrUpdate(indexName, list.Skip(j * bulkNum).Take(bulkNum).ToList())));
                }
                await Task.WhenAll(tasks);
            }
            return result;
        }

        public async Task<Tuple<List<string>, List<string>>> BulkAddOrUpdateDocumentV2Async(List<TAggregateRoot> list, int bulkNum = 5000)
        {
            var indexName = GetEnvIndexName();

            if (list.Count <= bulkNum)
            {
                return await BulkAddOrUpdateV2(indexName, list);
            }
            else
            {
                var total = (int)Math.Ceiling(list.Count * 1.0f / bulkNum);

                var tasks = new List<Task<Tuple<List<string>, List<string>>>>();
                for (var i = 0; i < total; i++)
                {
                    var j = i;
                    tasks.Add(
                        BulkAddOrUpdateV2(indexName, list.Skip(j * bulkNum).Take(bulkNum).ToList())
                        );
                }
                var taskResult = await Task.WhenAll(tasks);

                List<string> success = new List<string>();
                List<string> faild = new List<string>();

                foreach (var item in taskResult)
                {
                    if (item.Item1 != null)
                        success.AddIfNotContains(item.Item1);

                    if (item.Item2 != null)
                        faild.AddIfNotContains(item.Item2);
                }

                return new Tuple<List<string>, List<string>>(success, faild);
            }
        }

        public async Task BulkDeleteDocumentAsync(List<TAggregateRoot> list, int bulkNum = 1000)
        {
            var indexName = GetEnvIndexName();
            if (list.Count <= bulkNum)
            {
                await BulkDelete(indexName, list);
            }
            else
            {
                var total = (int)Math.Ceiling(list.Count * 1.0f / bulkNum);
                var tasks = new List<Task>();
                for (var i = 0; i < total; i++)
                {
                    var j = i;
                    tasks.Add(Task.Factory.StartNew(() =>
                        BulkDelete(indexName, list.Skip(j * bulkNum).Take(bulkNum).ToList())));
                }

                await Task.WhenAll(tasks);
            }
        }

        private async Task<string> BulkAddOrUpdate(string indexName, List<TAggregateRoot> list)
        {
            var result = string.Empty;
            var bulk = new BulkRequest(indexName)
            {
                Operations = new List<IBulkOperation>()
            };
            try
            {
                foreach (var item in list)
                {
                    if (item == null) continue;
                    var document = new BulkIndexOperation<TAggregateRoot>(item);
                    if (!string.IsNullOrEmpty(item.Routing))
                    {
                        document.Routing = item.Routing;
                    }
                    bulk.Operations.Add(document);
                }
                var response = await _esClient.BulkAsync(bulk);
                if (response.Errors)
                {
                    //result = string.Join(",", response.ItemsWithErrors.Select(i => i.Id));
                    result = $"BulkAddOrUpdate failed at index {indexName}:{JsonConvert.SerializeObject(response)}";
                }

                if (!response.IsValid)
                {
                    result = response.OriginalException.Message;
                    //result = $"Bulk AddOrUpdate failed at index {indexName}:{JsonConvert.SerializeObject(response)}";
                }
                else if (!response.ApiCall.Success)
                {
                    result = response.ApiCall.OriginalException.InnerException.Message;
                    //result = $"Bulk AddOrUpdate failed at index {indexName}:{JsonConvert.SerializeObject(response)}";
                }
            }
            catch (Exception ex)
            {
                result = $"BulkAddOrUpdate failed at index {indexName}:{ex.Message},{ex.StackTrace}";
            }
            return result;
        }

        private async Task<Tuple<List<string>, List<string>>> BulkAddOrUpdateV2(string indexName, List<TAggregateRoot> list)
        {

            var bulk = new BulkRequest(indexName)
            {
                Operations = new List<IBulkOperation>()
            };
            foreach (var item in list)
            {
                if (item == null) continue;
                var document = new BulkIndexOperation<TAggregateRoot>(item);
                if (!string.IsNullOrEmpty(item.Routing))
                {
                    document.Routing = item.Routing;
                }
                bulk.Operations.Add(document);
            }
            var response = await _esClient.BulkAsync(bulk);
            if (response.Errors)
            {
                return new Tuple<List<string>, List<string>>(response.Items.Select(i => i.Id).ToList(), response.ItemsWithErrors.Select(i => i.Id).ToList());
            }
            return new Tuple<List<string>, List<string>>(response.Items.Select(i => i.Id).ToList(), null);
        }

        private async Task BulkDelete(string indexName, List<TAggregateRoot> list)
        {
            var bulk = new BulkRequest(indexName)
            {
                Operations = new List<IBulkOperation>()
            };
            foreach (var item in list)
            {
                bulk.Operations.Add(new BulkDeleteOperation<TAggregateRoot>(new Id(item)));
            }

            var response = await _esClient.BulkAsync(bulk);
            if (response.Errors)
                throw new ElasticSearchException($"Bulk Delete Document at index {indexName} :{response.ServerError.Error.Reason}");
        }

        public async Task DeleteDocumentAsync(TAggregateRoot model)
        {
            var indexName = GetEnvIndexName();
            var response = await _esClient.DeleteAsync(new DeleteRequest(indexName, new Id(model)));
            if (response.ServerError == null) return;
            throw new Exception($"Delete Document at index {indexName} :{response.ServerError.Error.Reason}");
        }

        public async Task DeleteIndexAsync()
        {
            var indexName = GetEnvIndexName();
            var response = await _esClient.Indices.DeleteAsync(indexName);
            if (response.Acknowledged) return;
            throw new Exception($"Delete index {indexName}，failed :{response.ServerError.Error.Reason}");
        }

        public async Task DeleteIndexAsync(string indexName)
        {
            var response = await _esClient.Indices.DeleteAsync(indexName);
            if (response.IsValid) return;
            throw new Exception($"Delete index {indexName}，failed :{response.ServerError.Error.Reason}");
        }

        public async Task<List<string>> GetAllIndex()
        {
            List<string> index = new List<string>();
            var allIndexes = await _esClient.Cat.IndicesAsync();
            if (allIndexes.IsValid)
            {
                foreach (var item in allIndexes.Records)
                {
                    index.Add(item.Index);
                }
            }

            return index;
        }

        public async Task ReIndex()
        {
            await DeleteIndexAsync();
            await CreateIndexAsync();
        }

        public async Task ReBuildDocumentAsync(int shard = 5, int numberOfReplicas = 1)
        {
            var indexName = GetEnvIndexName();
            var result = await _esClient.Indices.GetAliasAsync(indexName);
            var oldName = result.Indices.Keys.First();

            //创建新的索引
            var newIndex = $"{indexName}_{DateTime.Now.Ticks}";
            var createResult = await _esClient.Indices.CreateAsync(newIndex,
                c =>
                    c.Index(newIndex).Settings(
                            o =>
                            o.NumberOfShards(shard)
                             .NumberOfReplicas(numberOfReplicas)
                             .Setting("max_result_window", int.MaxValue))
                        .Map(m => m.AutoMap<TAggregateRoot>()));

            if (!createResult.Acknowledged)
            {
                throw new Exception($"reBuild create newIndex {indexName}，failed :{result.ServerError.Error.Reason}");
            }

            //重建索引数据
            var reResult = await _esClient.ReindexOnServerAsync(descriptor => descriptor
                .Source(source => source.Index(indexName))
                .Destination(dest => dest.Index(newIndex)));

            if (reResult.ServerError != null)
            {
                throw new Exception($"reBuild {indexName} data，failed :{reResult.ServerError.Error.Reason}");
            }

            //删除旧索引
            var deleteResult = await _esClient.Indices.DeleteAsync(oldName);
            if (!deleteResult.Acknowledged)
            {
                throw new Exception($"reBuild delete old Index {oldName.Name}，failed :{deleteResult.ServerError.Error.Reason}");
            }

            //指定索引别名
            var reAliasResult = await _esClient.Indices.PutAliasAsync(newIndex, indexName);
            if (!reAliasResult.IsValid)
            {
                throw new Exception($"reBuild set Alias {indexName}，failed :{reAliasResult.ServerError.Error.Reason}");
            }
        }

        /// <summary>
        /// 查询索引数据
        /// </summary>
        /// <param name="query"></param>
        /// <param name="trackTotalHits"></param>
        /// <param name="indexName">若传递的参数indexName为空，则自动取GetEnvIndexName()</param>
        /// <returns></returns>
        public async Task<ISearchResponse<TAggregateRoot>> Search(SearchDescriptor<TAggregateRoot> query, bool trackTotalHits = false, string indexName = "")
        {
            if (indexName.IsNullOrWhiteSpace())
            {
                indexName = GetEnvIndexName();
            }

            query.Index(indexName);
            if (trackTotalHits)
            {
                query.TrackTotalHits(trackTotalHits);
            }
#if DEBUG
            var json = _esClient.ToRawRequest(query);
            _logger.LogWarning($"ES同步客户json:{json}");
#endif
            var response = await _esClient.SearchAsync<TAggregateRoot>(query);

            if (!response.IsValid)
                throw new Exception(response.DebugInformation, response.OriginalException);

            return response;
        }

        /// <summary>
        /// 获取查询语句
        /// </summary>
        /// <param name="query"></param>
        /// <param name="trackTotalHits"></param>
        /// <returns></returns>
        public string GetSearchTxt(SearchDescriptor<TAggregateRoot> query, bool trackTotalHits = false)
        {
            var indexName = GetEnvIndexName();

            query.Index(indexName);
            if (trackTotalHits)
            {
                query.TrackTotalHits(trackTotalHits);
            }

            return _esClient.ToRawRequest(query);
        }

        public async Task<CountResponse> Count(CountDescriptor<TAggregateRoot> query)
        {
            var indexName = GetEnvIndexName();

            query.Index(indexName);

            var response = await _esClient.CountAsync(query);

            if (!response.IsValid)
                throw new Exception(response.DebugInformation, response.OriginalException);

            return response;
        }

        public async Task<ISearchResponse<TAggregateRoot>> SearchAsync(SearchDescriptor<TAggregateRoot> query, int skip, int size,
            string[] includeFields = null, string preTags = "<strong style=\"color: red;\">", string postTags = "</strong>", bool disableHigh = false, params string[] highField)
        {
            var indexName = GetEnvIndexName();

            query.Index(indexName);

            var highlight = new HighlightDescriptor<TAggregateRoot>();
            if (disableHigh)
            {
                preTags = "";
                postTags = "";
            }

            highlight.PreTags(preTags).PostTags(postTags);

            var isHigh = highField != null && highField.Length > 0;

            var hfs = new List<Func<HighlightFieldDescriptor<TAggregateRoot>, IHighlightField>>();

            //分页
            query.Skip(skip).Take(size);
            //关键词高亮
            if (isHigh)
            {
                foreach (var s in highField)
                {
                    hfs.Add(f => f.Field(s));
                }
            }

            highlight.Fields(hfs.ToArray());
            query.Highlight(h => highlight);

            if (includeFields != null)
            {
                query.Source(ss => ss.Includes(ff => ff.Fields(includeFields.ToArray())));
            }

            var response = await _esClient.SearchAsync<TAggregateRoot>(query);
            return response;
        }

        public async Task<CountResponse> CountAsync(Func<QueryContainerDescriptor<TAggregateRoot>, QueryContainer> query)
        {
            var indexName = GetEnvIndexName();
            var response = await _esClient.CountAsync<TAggregateRoot>(c => c.Index(indexName).Query(query));
            return response;
        }
    }
}
